package com.atguigu.flink.datastreamapi.transform;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Created by Smexy on 2022/12/13
 *
 *  在sql中，不group by是无法调用聚合函数。
 *
 *  在代码中，不分组(keyBy)，无法调用聚合的算子。
 *
 *  ------------------
 *      按照传感器的id，统计不同种类传感器的vc和
 *          sum:  对指定的字段进行累加，其他字段取的都是组内第一条数据的属性。
 *          max|min:  对指定的字段取最大和最小值，其他字段取的都是组内最大或最小数据的属性。
 *          maxBy|minBy:  对指定的字段取最大和最小值，所有字段取的都是组内最大或最小数据的属性。
 *                  minBy(字段)： 遇到两条同样是最小的，依旧取第一条的所有属性
 */
public class Demo7_Agg
{
    public static void main(String[] args) {


       StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

               env
                  .socketTextStream("hadoop103", 8888)
                  .map(new WaterSensorMapFunction())
                  .keyBy(WaterSensor::getId)
                  //.sum("vc")
                  //.max()
                  //.min("vc")
                  // 如果遇到两条数据的vc都是最小的，默认不取第一条，取最后一条
                  .minBy("vc",false)
                  .print();


               try {
                           env.execute();
                       } catch (Exception e) {
                           e.printStackTrace();
                       }

    }
}
